package com.lagou.mr.sortnumber;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.concurrent.atomic.AtomicLong;

public class SortNumberReducer extends Reducer<LongWritable, NullWritable, OrderLong, NullWritable> {

    /**
     * 由于 ReduceTask 肯定只有一个，所以这里可以使用 java 原子类来作为递增计数器
     */
    private static final AtomicLong INC = new AtomicLong(0);

    @Override
    protected void reduce(LongWritable key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
        // 利用mr的shuffle对数据进行了排序
        // 为了避免LongWritable.compareTo方法导致数字相等就被当成对象相等，而合并了key，
        // 所以遍历value，获取隐藏key
        for (NullWritable value : values) {
            long order = INC.incrementAndGet();
            OrderLong orderLong = new OrderLong(order, key.get());
            context.write(orderLong, value);
        }
    }

}
